package main.test.GraphFramesAPI

import org.apache.spark.sql.functions.{col, lit, when}
import org.apache.spark.sql.{Column, DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.graphframes.GraphFrame

object MotifFinding {
  def main(args: Array[String]): Unit = {
    //  def constructGraph: GraphFrame = {
    //    The following example demonstrates how
    //    to create a GraphFrame from vertex and edge DataFrame
    val sparkConf = new SparkConf()
    sparkConf.setAppName("createGraph").setMaster("local[*]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)
    val v = sqlContext.createDataFrame(List(
      ("a", "Alice", 34),
      ("b", "Bob", 36),
      ("c", "Charlie", 30),
      ("d", "David", 29),
      ("e", "Esther", 32),
      ("f", "Fanny", 36),
      ("g", "Gabby", 60)
    )).toDF("id", "name", "age")

    // Edge DataFrame
    val e = sqlContext.createDataFrame(List(
      ("a", "b", "friend"),
      ("b", "c", "follow"),
      ("c", "b", "follow"),
      ("f", "c", "follow"),
      ("e", "f", "follow"),
      ("e", "d", "friend"),
      ("d", "a", "friend"),
      ("a", "e", "friend")
    )).toDF("src", "dst", "relationship")


    // Create a GraphFrame
    val g = GraphFrame(v, e)
    //    Motif finding refers to
    //    searching for structural patterns
    //    in a graph.
    // details
    /**
     * GraphFrame motif finding
     * uses a simple Domain-Specific Language (DSL)
     * for expressing structural queries.
     *
     * DSL for expressing structural patterns:
     * 1 The basic unit of a pattern is an edge
     * "(a)-[e]->(b)" expresses an edge e from vertex a to vertex b
     * 2 A pattern is expressed as a union of edges
     * Motif "(a)-[e]->(b); (b)-[e2]->(c)" specifies two edges from a to b to c.
     *
     * 3 Within a pattern, names can be assigned to vertices and edges.
     * "(a)-[e]->(b)" has three named elements: vertices a,b and edge e. These names serve two purposes:
     * 1) The names can identify common elements among edges
     * "(a)-[e]->(b); (b)-[e2]->(c)" specifies that the same vertex b is the destination of edge e and source of edge e2.
     * 2) The names are used as column names in the result DataFrame.
     * 3) Be aware that names do not identify distinct elements
     * in the motif "(a)-[e]->(b); (b)-[e2]->(c)", the names a and c could refer to the same vertex
     * To restrict named elements to be distinct vertices or edges,
     * use post-hoc filters such as resultDataframe.filter("a.id != c.id")
     *
     * 4  omit names for vertices or edges in motifs when not needed.
     * E.g., "(a)-[]->(b)" expresses an edge between vertices a,b but does not assign a name to the edge.
     * omit names for vertices or edges in motifs when not needed. E.g., "(a)-[]->(b)" expresses an edge between vertices a,b but does not assign a name to the edge.
     * 5 An edge can be negated to indicate that the edge should not be present in the graph.
     * "(a)-[]->(b); !(b)-[]->(a)" finds edges from a to b for which there is no edge from b to a.
     * Logical operator not !!!!
     * Restrictions:
     * 1 Motifs are not allowed to contain edges without any named elements: "()-[]->()" and "!()-[]->()" are prohibited terms.
     * 2 Motifs are not allowed to contain named edges within negated terms (since these named edges would never appear within results). E.g., "
     * !(a)-[ab]->(b)" is invalid, but "!(a)-[]->(b)" is valid.
     *
     *
     *
     */
    val motifs: DataFrame = g.find("(a)-[e]->(b); (b)-[e2]->(a)")
    motifs.show()

    // More complex queries can be expressed by applying filters.
    motifs.filter("b.age > 30").show()

    /**
     * The next examples demonstrate more complex queries
     * which carry state along a path in the motif.
     * ==== combining GraphFrame motif finding with filters on the result, where the filters use sequence operations to construct a series of DataFrame Columns.====
     * suppose one wishes to identify a chain of 4 vertices with some property defined by a sequence of functions. That is, among chains of 4 vertices a->b->c->d, identify the subset of chains matching this complex filter:
     * Initialize state on path.
     * Update state based on vertex a.
     * Update state based on vertex b.
     * Etc. for c and d.
     * If final state matches some condition, then the chain is accepted by the filter.
     * we identify chains of 4 vertices such that at least 2 of the 3 edges are “friend” relationships.
     * the state is the current count of “friend” edges
     * it could be any DataFrame column.
     */

    val chain4: DataFrame = g.find("" +
      "(a)-[ab]->(b); " +
      "(b)-[bc]->(c); " +
      "(c)-[cd]->(d)")


    // Query on sequence, with state (cnt)
    //  (a) Define method for updating state given the next element of the motif.
    def sumFriends(cnt: Column, relationship: Column): Column = {
      when(relationship === "friend", cnt + 1).otherwise(cnt)

    }

    //  (b) Use sequence operation to apply method to sequence of elements in motif.
    //      In this case, the elements are the 3 edges.
    val condition: Column = {
      Seq("ab", "bc", "cd")
        .foldLeft(lit(0))((cnt, e) => sumFriends(cnt, col(e)("relationship")))
    }


    // (c) Apply filter to DataFrame
    val chainWith2Friends2 = chain4.where(condition >= 2)

    chainWith2Friends2.show()


  }

}
